
package org.codebase.hadoop.zookeeper.base;

import java.util.Map;

import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.ZooKeeper;

/**
 * 节点监听类
 **************************************************
 * ClassName:DataWatcher

 * @author:xiaozhichao
 * @date:2018年8月15日
 **************************************************
 */
public class DataWatcher implements Watcher, Runnable {

	private String hostPort;
	@SuppressWarnings("unused")
	private String zkDatePath;
	private ZooKeeper zk;
	private Map<Integer,Handler> handlers;
	
	public DataWatcher(String hostPort,String zkDatePath,Map<Integer,Handler> handlers){
		this.hostPort = hostPort;
		this.zkDatePath = zkDatePath;
		this.handlers = handlers;
		try {
			zk = new ZooKeeper(this.hostPort,2000,this);
			if (zk.exists(zkDatePath, this) == null) {
	            zk.create(zkDatePath, "".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
	        }
		} catch (Exception e) {
			e.printStackTrace();
		}
	}
	
	@Override
	public void process(WatchedEvent event) {
		Integer type = event.getType().getIntValue();
		Handler handler = this.handlers.get(type);
		if(handler!=null){
			handler.handle(this.zk,this);
		}
	}

	@Override
	public void run() {
		try{
			synchronized(this){
				while(true){
					wait();
				}
			}
		}catch(Exception e){
			e.printStackTrace();
			Thread.currentThread().interrupt();
		}
	}

}
